package com.meihua.code.leader;



import ch.qos.logback.core.util.TimeUtil;
import com.alibaba.fastjson.JSONObject;
import com.meihua.code.leader.domain.HttpRequest;
import com.meihua.code.leader.domain.HttpResponse;
import com.meihua.code.leader.domain.Node;
import com.meihua.code.leader.domain.NodeRoleEnum;
import com.meihua.code.util.PropertiesUtils;
import com.meihua.code.util.TimeUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.net.*;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;


/**
 *
 * @author humeihua
 * @date 2021/08/10
 */
public class RaftNode {

    private transient final  static Logger logger= LoggerFactory.getLogger(Raft.class);
    /**
     * 节点角色
     **/
    private NodeRoleEnum role;

    private  String serverAddress;

    private final int port;

    /**
     * 节点名称
     */
    private String nodeName;

    /**
     * 心跳超时时间  随机 用来减少竞争
     **/
    private final long beatTimeOut;

    /**
     * 上次心跳时间
     **/
    private  long lastBeatTime =System.currentTimeMillis();
    /**
     * 任期 选举轮数
     **/
    private final AtomicInteger term =new AtomicInteger(0);

    /**
     * leader 节点的信息
     **/
    private RaftNode leader;

    /**
     * 日志index 对比使用
     **/
    private final AtomicInteger logIndex =new AtomicInteger(0);

    /**
     * 其他节点 模拟交互通信
     **/
    private transient List<Node> nodes;

    /**
     * 投票记录
     */
    private transient final CopyOnWriteArraySet<Integer> VOTE_MAP = new CopyOnWriteArraySet<>();

    /**
     * 日志复制队列
     **/
    private final  LinkedBlockingDeque BIN_LOG=new LinkedBlockingDeque<Object>();


    private transient ScheduledExecutorService executor ;

    public RaftNode(String nodeName,int port) {
        this.role = NodeRoleEnum.follower;
        //心跳超时时间
        this.beatTimeOut = new Random().nextInt(1000 - 500 + 1) + 500;
        this.port=port;
        try {
            InetAddress addr  = InetAddress.getLocalHost();
            this.serverAddress=addr.getHostAddress();
            this.nodeName = nodeName;
            this.nodes = getNodes(addr.getHostAddress(),port);
        } catch (UnknownHostException e) {
           logger.error("error get ip :",e);
        }
    }

    public static List<Node> getNodes(String localAddress,int port){
        List<Node> list= new ArrayList<>(3);
        String cluster = PropertiesUtils.getConfig("raft.cluster.config");
        String[] ipConfigs = cluster.split(",");
        int i=1;
        for (String ip:ipConfigs){
                ip = ip.startsWith("http://") || ip.startsWith("https://")? ip: "http://"+ip;
                URI uri = URI.create(ip);
                //如果是自己 排除
                if (!uri.getHost().equals(localAddress)||uri.getPort()!=port){
                    list.add(new Node("节点："+ i++,uri.getHost(),uri.getPort()));
                }
        }
        logger.info("cluster config :{} ",list);
        return list;
    }

    public boolean commonNotify(AtomicInteger success,Function<Node,Boolean> function){
        CountDownLatch countDownLatch = new CountDownLatch(nodes.size());
        for(Node node:nodes){
            try {
                executor.execute(()->{
                    try {
                        //执行对应逻辑
                        if (function.apply(node)){
                            success.incrementAndGet();
                        }
                    }catch (Exception e){
                        logger.error("{} Notify error ",this,e);
                    }finally {
                        countDownLatch.countDown();
                    }
                });
            }catch (Exception e){
                logger.error(e.getMessage());
            }
        }
        try {
            countDownLatch.await(1000,TimeUnit.MILLISECONDS);
        } catch (InterruptedException e) {
            logger.error("{} time out ",this);
            return false;
        }
        //票数超过半数 返回true
         return success.get()>(this.nodes.size()+1)/2.0;
    }

    public boolean proVote(){
       AtomicInteger success = new AtomicInteger(1);
       return commonNotify(success,(node)->{
           HttpResponse rsp = sendRequest(UrlEnum.attempt_vote.getUrl(), node, this.nodeName+"尝试锁定");
            // logger.debug("resp :{}",rsp);
           return checkTerm(rsp) && rsp.getCode()==200&&this.term.get()>=rsp.getTerm()&&this.logIndex.get()>=rsp.getLogIndex();
       });
    }

    public boolean  voteRequest(){
        //随机的休眠时间 减少竞争
        //sleep();
        //休眠期间可能会存在其他节点成为leader
        if (this.role==NodeRoleEnum.candidate) {
            //向其他节点发送投票请求
            AtomicInteger success = new AtomicInteger(0);
            //自己投自己一票
            if (VOTE_MAP.add(this.term.get())) {
                success.incrementAndGet();
            }
            return commonNotify(success, (node) -> {
                HttpResponse<Boolean> rsp = sendRequest(UrlEnum.vote_request.getUrl(), node, this.nodeName+"投票请求");
                // logger.debug("resp :{}", rsp);
                return checkTerm(rsp) && rsp.getCode() == 200 && rsp.getBody();
            });
        }
        return false;
    }

    public boolean notifyNode(){
        //向其他节点发送leader 通知
        AtomicInteger success=new AtomicInteger(1);
        return commonNotify(success, (node) -> {
            HttpResponse<Boolean> rsp = sendRequest(UrlEnum.leader_notify.getUrl(), node, this.nodeName+"成为leader");
            //  logger.debug("resp :{}", rsp);
            return checkTerm(rsp) && rsp.getCode() == 200 &&rsp.getBody();
        });
    }


    @SuppressWarnings("InfiniteLoopStatement")
    public void start(){
        if (this.executor!=null){
            this.executor.shutdown();
        }
        this.executor = new ScheduledThreadPoolExecutor(5+nodes.size());
        //心跳超时线程
        this.executor.scheduleAtFixedRate(()->{
            //检测到心跳超时
            if (checkBeatTimedOut()&&this.role!=NodeRoleEnum.Leader){
                //sleep();
                logger.info("{} beat time out  last beat time :{}",this,TimeUtils.getStringDateShort(lastBeatTime));
                //proVote 流程
                if (!checkBeatTimedOut()){
                    logger.info("心跳恢复 休眠期间产生了leader");
                    return;
                }
                if (!proVote()){
                   return;
                }
                if (!checkBeatTimedOut()){
                    logger.info("心跳恢复 准备期间产生了leader");
                    return;
                }
                    //任期+1
                    this.term.incrementAndGet();
                    this.role=NodeRoleEnum.candidate;
                    //触发选举 选举通过
                    if (voteRequest()){
                        //发送成为leader 通知
                        if (notifyNode()){
                            //成为leader
                            logger.info("{} 成为leader",this);
                            this.role=NodeRoleEnum.Leader;
                        }else {
                            //成为leader 失败 期待下次的选举
                            logger.info("{} 成为leader 失败 期待下次的选举",this);
                            this.role=NodeRoleEnum.follower;
                        }
                    }else {

                    }
            }
        },0,100,TimeUnit.MILLISECONDS);

        //leader 心跳 每500ms 一次通知
        this.executor.scheduleAtFixedRate(()->{
            if(this.role==NodeRoleEnum.Leader){
                AtomicInteger success=new AtomicInteger(1);
               if(commonNotify(success,(node)->{
                    HttpResponse<Boolean> rsp = sendRequest(UrlEnum.beat_notify.getUrl(), node, this.nodeName+"的心跳");
                    return checkTerm(rsp) && rsp.getCode() == 200 && rsp.getBody();
                })){
                   this.lastBeatTime=System.currentTimeMillis();
                }
            }
        },0,200,TimeUnit.MILLISECONDS);

        executor.execute(()->{
            try {
                ServerSocket socket = new ServerSocket(port);
                //socket 线程 与其他节点交互
                logger.info("{} start !",this);
                while (true) {
                    Socket clientSocket = socket.accept();
                    //logger.info("{} 新的链接 :{}",this,clientSocket.getInetAddress());
                    //开户启一个新线程处理新的连接请求，一个线程可以处理同一条连接上的多个请求，而不是每个请求创建一个连接
                    executor.execute(()->{
                        try {
                            DataInputStream br = new DataInputStream(clientSocket.getInputStream());
                            String request =  br.readUTF() ;
                            HttpRequest requestBody = JSONObject.parseObject(request, HttpRequest.class);
                            HttpResponse response =  requestEventHandler(requestBody);
                            new DataOutputStream(clientSocket.getOutputStream()).writeUTF( JSONObject.toJSONString(response));
                            clientSocket.close();
                        } catch (Exception e) {
                            e.printStackTrace();
                        } finally {
                            try {
                                clientSocket.close();
                            } catch (IOException e1) {
                                e1.printStackTrace();
                            }
                        }
                    });
                }
            } catch (IOException e) {
                logger.error("启动http 服务器失败 地址：{} 节点名称：{}",serverAddress,nodeName);
                System.exit(1);
            }
        });
    }

    public void sleep(long ms){
        try {
            Thread.sleep(ms);
        }catch (InterruptedException e){
            logger.error("",e);
        }
    }

    public <R> HttpResponse<R> sendRequest(String url,Node node,Object body){
        DataOutputStream pw =null;
        Socket socket=null;
        try {
             socket = new Socket(node.getIp(), node.getPort());
            pw=new DataOutputStream(socket.getOutputStream());
            pw.writeUTF( JSONObject.toJSONString(new HttpRequest(url,this,body,term.get(),logIndex.get())));
            pw.flush();
            //读取服务器返回的消息
            DataInputStream br = new DataInputStream(socket.getInputStream());
            //读取客户端发送来的消息
            HttpResponse response = JSONObject.parseObject(br.readUTF(),HttpResponse.class) ;
            socket.close();
           // logger.debug(" response {}",response.toString());
            return   response;
        } catch (IOException e) {
            logger.error(e.getMessage());
            return new HttpResponse(500,"request IP "+node +e.getMessage(),-1,-1);
        }
    }

    public int sleep(){
        try {
            int ms = (int)(Math.random()*8000);
            Thread.sleep(ms);
            return   ms;
        }catch (InterruptedException e){
            logger.error("",e);
        }
        return -1;
    }

    public boolean checkBeatTimedOut(){
       return System.currentTimeMillis()-lastBeatTime>beatTimeOut;
    }

    public boolean checkTerm(HttpResponse response){
        if (this.term.get()<response.getTerm()){
            //出现比自己大的term
            this.role=NodeRoleEnum.follower;
            return false;
        }
        return true;
    }

    /**
     * 请求事件处理
     **/
    public <T> HttpResponse requestEventHandler(HttpRequest<T> request){
        logger.info("{}  {}",this,request);
        switch (UrlEnum.getUrlEnum(request.getUrl())){
            case leader_notify:
                //leader 的term比自己低 拒绝
                 if (this.term.get()>request.getTerm()){
                     return  new HttpResponse<>(200,false,this.term.get(),this.logIndex.get());
                 }
                 this.leader=request.getNode();
                 //变更为follower
                 this.role=NodeRoleEnum.follower;
                 this.term.set(request.getTerm());
                 //接收来自于Leader的日志
                 return new HttpResponse<>(200,true,this.term.get(),this.logIndex.get());
            case vote_request:
                //leader 的term比自己低 或已投票过  拒绝
                if (this.term.get()>request.getTerm()|| !this.VOTE_MAP.add(request.getTerm())){
                    return  new HttpResponse<>(200,false,this.term.get(),this.logIndex.get());
                }
                //任期变更
                this.term.set(request.getTerm());
                return new HttpResponse<>(200,true,this.term.get(),this.logIndex.get());
            case attempt_vote:
                //尝试投票
                if (this.term.get()>request.getTerm()||this.logIndex.get()>request.getLogIndex()){
                    return  new HttpResponse<>(200,false,this.term.get(),this.logIndex.get());
                }
                return  new HttpResponse<>(200,true,this.term.get(),this.logIndex.get());
            case beat_notify:
                if (this.term.get()>request.getTerm()){
                    return  new HttpResponse<>(200,false,this.term.get(),this.logIndex.get());
                }
               // logger.info("{} 收到leader 的心跳",this);
                if (this.role==NodeRoleEnum.candidate){
                    this.role=NodeRoleEnum.follower;
                }
                this.lastBeatTime=System.currentTimeMillis();
                this.term.set(request.getTerm());
                return  new HttpResponse<>(200,true,this.term.get(),this.logIndex.get());
            case log_notify:

            case unknown:
            default:
               return new HttpResponse<>(404,false,this.term.get(),this.logIndex.get());
        }
    }


    /**
     * url枚举
     *
     * @author humeihua
     * @date 2021/08/17
     */
    enum UrlEnum {

        /**
         * 未知的
         */
        unknown(-1,"unknown",null,null),

        /**
         * leader 通知
         */
        leader_notify(1,"leader_notify",Boolean.class,Boolean.class),
        /**
         * 投票要求
         */
        vote_request(2,"vote_request",Boolean.class,Boolean.class),
        /**
         * 尝试
         */
        attempt_vote(3,"attempt_vote",Boolean.class,Boolean.class),
        /**
         * 日志通知
         */
        log_notify(4,"notify_all",Boolean.class,Boolean.class),

        beat_notify(5,"beat_notify",Boolean.class,Boolean.class),
        /**
         * 客户端事件
         */
        client_event(6,"client_event",Boolean.class,Boolean.class)
        ;

        /**
         * 指数
         */
        private  int index;
        /**
         * url
         */
        private  String url;
        /**
         * 请求体
         */
        private  Class RequestBody;
        /**
         * 响应体
         */
        private  Class ResponseBody;

        UrlEnum(int index, String url, Class requestBody, Class responseBody) {
            this.index = index;
            this.url = url;
            RequestBody = requestBody;
            ResponseBody = responseBody;
        }

        public static UrlEnum getUrlEnum(String url){
            for(UrlEnum urlEnum:UrlEnum.values()){
                if (urlEnum.url.equals(url)){
                    return urlEnum;
                }
            }
            return unknown;
        }

        public String getUrl() {
            return url;
        }
    }

    @Override
    public String toString() {
        return nodeName +
                " " + role +
                " term:" + term +
                " leader:" + (leader != null ?  leader.nodeName:null )+
                " log:" + logIndex;
    }


}
